package hadoop.test;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.*;

public class WordCountReducer extends Reducer<Text, IntWritable, Text, LongWritable> {

    private static Map<String, Integer> map = new HashMap<String, Integer>();

    @Override
    /*
     * reduce方法提供给reduce task进程来调用
     *
     * reduce task会将shuffle阶段分发过来的大量kv数据对进行聚合，聚合的机制是相同key的kv对聚合为一组
     * 然后reduce task对每一组聚合kv调用一次我们自定义的reduce方法
     * 比如：<hello,1><hello,1><hello,1><tom,1><tom,1><tom,1>
     *  hello组会调用一次reduce方法进行处理，tom组也会调用一次reduce方法进行处理
     *  调用时传递的参数：
     *          key：一组kv中的key
     *          values：一组kv中所有value的迭代器
     */
    protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {
        //定义一个计数器
        int count = 0;
        //通过value这个迭代器，遍历这一组kv中所有的value，进行累加
        for (IntWritable value : values) {
            count += value.get();
        }

        //保存这个单词的统计结果
        map.put(key.toString(), count);
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        Map<String, Integer> resultMap = sortMapByValue(map);

        long wordCount = 0;  //统计总数字

        for (Map.Entry<String, Integer> entry : resultMap.entrySet()) {
            wordCount += entry.getKey().length()*entry.getValue();  //长度*出现次数
            context.write(new Text(entry.getKey()), new LongWritable(entry.getValue()));

        }

        context.write(new Text("AllCount"), new LongWritable(wordCount));


        super.cleanup(context);
    }

    public static Map<String, Integer> sortMapByValue(Map<String, Integer> oriMap) {
        if (oriMap == null || oriMap.isEmpty()) {
            return null;
        }
        Map<String, Integer> sortedMap = new LinkedHashMap<String, Integer>();
        List<Map.Entry<String, Integer>> entryList = new ArrayList<Map.Entry<String, Integer>>(
                oriMap.entrySet());
        Collections.sort(entryList, new MapValueComparator());

        Iterator<Map.Entry<String, Integer>> iter = entryList.iterator();
        Map.Entry<String, Integer> tmpEntry = null;
        while (iter.hasNext()) {
            tmpEntry = iter.next();
            sortedMap.put(tmpEntry.getKey(), tmpEntry.getValue());
        }
        return sortedMap;
    }
}
